package com.sec.etech.redis.stream;

import com.sec.etech.redis.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class RedisStreamUtil {
    @Resource
    //private StringRedisTemplate redisTemplate;
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private RedisUtil redisUtil;

    /**
     * 创建消费组
     *
     * @param key
     * @param group
     * @return
     */
    public String creatGroup(String key, String group) {
        return redisTemplate.opsForStream().createGroup(key, group);
    }

    /**
     * 创建消费组，如果不存在创建stream
     *
     * @param key
     * @param group
     * @param offset
     * @return
     */
    public String creatGroup(String key, ReadOffset offset, String group) {
        return redisTemplate.opsForStream().createGroup(key, offset, group);
    }

    /**
     * 创建消费组，如果不存在创建stream
     *
     * @param key
     * @param group
     * @return
     */
    public String creatGroupMKStream(String key, String group) {
        return redisTemplate.opsForStream().createGroup(key, ReadOffset.latest(), group);
    }

    /**
     * 消费组的消费者信息
     *
     * @param key
     * @param group
     * @return
     */
    public StreamInfo.XInfoConsumers consumers(String key, String group) {
        return redisTemplate.opsForStream().consumers(key, group);
    }

    /**
     * 获取消费组
     *
     * @param key
     * @return
     */
    public StreamInfo.XInfoGroups groups(String key) {
        return redisTemplate.opsForStream().groups(key);
    }

    /**
     * 确认已消费
     *
     * @param key
     * @param group
     * @param recordIds
     * @return
     */
    public Long ack(String key, String group, String... recordIds) {
        return redisTemplate.opsForStream().acknowledge(key, group, recordIds);
    }

    /**
     * 追加消息
     *
     * @param key
     * @param field
     * @param value
     * @return
     */
    public String add(String key, String field, Object value) {
        Map<String, Object> content = new ConcurrentHashMap<>(1);
        content.put(field, value);
        return add(key, content);
    }

    /**
     * 追加消息
     *
     * @param key     stream key
     * @param content 消息体（map结构）
     * @return
     */
    public String add(String key, Map<String, Object> content) {
        return redisTemplate.opsForStream().add(key, content).getValue();
    }

    /**
     * 追加消息
     *
     * @param key      stream key
     * @param recordId 自定义唯一消息ID
     * @param content  消息体（map结构）
     * @return
     */
    public String add(String key, Long recordId, Map<String, Object> content) {
        return redisTemplate.opsForStream().add(MapRecord.create(key, content)
                .withId(RecordId.of(System.currentTimeMillis(), recordId))).getValue();
    }

    /**
     * 裁剪stream消息数量（释放已消费消息的空间）
     *
     * @param key   stream key
     * @param count 数量
     * @return
     */
    public Long trim(String key, long count) {
        return redisTemplate.opsForStream().trim(key, count, false);
    }

    /**
     * 裁剪stream消息数量（释放已消费消息的空间）
     *
     * @param key                 stream key
     * @param count               数量
     * @param approximateTrimming 近似修剪
     * @return
     */
    public Long trim(String key, long count, boolean approximateTrimming) {
        return redisTemplate.opsForStream().trim(key, count, approximateTrimming);
    }

    /**
     * 删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度
     * 消息存储在stream的节点下，删除时仅对消息做删除标记，当一个节点下的所有条目都被标记为删除时，销毁节点
     *
     * @param key
     * @param recordIds
     * @return
     */
    public Long del(String key, String... recordIds) {
        return redisTemplate.opsForStream().delete(key, recordIds);
    }

    /**
     * 消息长度
     *
     * @param key
     * @return
     */
    public Long len(String key) {
        return redisTemplate.opsForStream().size(key);
    }

    /**
     * 单独消费：从开始读
     *
     * @param key
     * @return
     */
    public List<MapRecord<String, Object, Object>> read(String key) {
        return redisTemplate.opsForStream().read(StreamOffset.fromStart(key));
    }

    /**
     * 单独消费：从指定的ID开始读
     *
     * @param key
     * @param recordId
     * @return
     */
    public List<MapRecord<String, Object, Object>> read(String key, String recordId) {
        return redisTemplate.opsForStream().read(StreamOffset.from(MapRecord.create(key, new ConcurrentHashMap<>(1)).withId(RecordId.of(recordId))));
    }

    /**
     * 分组消费：从指定的ID开始读（读不到阻塞1秒）
     *
     * @param streamKey    stream key
     * @param group        组
     * @param consumerName 消费者名称
     * @param recordId     消息id
     * @return
     */
    public List<MapRecord<String, Object, Object>> readById(String streamKey, String group, String consumerName, String recordId) {
        return redisTemplate.opsForStream().read(Consumer.from(group, consumerName),
                StreamReadOptions.empty().count(1).block(Duration.ofSeconds(1)),
                StreamOffset.create(streamKey, ReadOffset.from(recordId)));
    }

    /**
     * 分组消费：从最新pending-list未读开始读（读不到阻塞1秒）
     *
     * @param streamKey    stream key
     * @param group        组
     * @param consumerName 消费者名称
     * @return
     */
    public List<MapRecord<String, Object, Object>> readLastConsumed(String streamKey, String group, String consumerName) {
        return redisTemplate.opsForStream().read(Consumer.from(group, consumerName),
                StreamReadOptions.empty().count(1)/*.block(Duration.ofSeconds(1))*/,
                StreamOffset.create(streamKey, ReadOffset.lastConsumed()));
    }

    /**
     * 分组消费：读所有未ack的信息（读不到阻塞1秒）
     *
     * @param streamKey stream key
     * @param group     组
     * @return
     */
    public PendingMessagesSummary readPendingList(String streamKey, String group) {
        return redisTemplate.opsForStream().pending(streamKey, group);
    }

    /**
     * 分组消费：读所有未ack的信息（读不到阻塞1秒）
     *
     * @param streamKey    stream key
     * @param group        组
     * @param consumerName 消费者名称
     * @return
     */
    public PendingMessages readPendingListId(String streamKey, String group, String consumerName) {
        return redisTemplate.opsForStream().pending(streamKey, Consumer.from(group, consumerName));
    }

    /**
     * 分组消费：读5条未ack的具体信息（读不到阻塞3秒）
     *
     * @param streamKey    stream key
     * @param group        组
     * @param consumerName 消费者名称
     * @return
     */
    public List<MapRecord<String, Object, Object>> readPendingList(String streamKey, String group, String consumerName) {
        return redisTemplate.opsForStream().read(Consumer.from(group, consumerName),
                StreamReadOptions.empty().noack().count(5).block(Duration.ofSeconds(3)),
                StreamOffset.create(streamKey, ReadOffset.from("0")));
    }

    /**
     * 检查并生成stream和group，默认从头消费
     */
    public void initGroup(String streamKey, String group) {
        this.initGroup(streamKey, ReadOffset.from("0-0"), group);
    }

    /**
     * 检查并生成stream和group
     */
    public void initGroup(String streamKey, ReadOffset readOffset, String group) {
        //检查stream key是否存在，不存在创建stream和group
        //ReadOffset readOffset = ReadOffset.from(offset);
        if (!redisUtil.hasKey(streamKey)) {
            this.creatGroup(streamKey, readOffset, group);
            return;
            //throw new RedisSystemException("stream key not exist error", null);
        }
        StreamInfo.XInfoGroups infoGroups = null;
        try {
            // 获取Stream的所有组信息
            infoGroups = this.groups(streamKey);
        } catch (RedisSystemException | InvalidDataAccessApiUsageException ex) {
            log.error("stream key not exist or commend error", ex);
        }
        // 遍历校验消费组是否存在
        boolean flag = false;
        Iterator<StreamInfo.XInfoGroup> it = infoGroups.iterator();
        while (it.hasNext()) {
            StreamInfo.XInfoGroup g = it.next();
            if (g.groupName().equals(group)) {
                flag = true;
            }
        }
        //如果不存在创建消费组
        if (!flag) {
            this.creatGroup(streamKey, readOffset, group);
        }
    }
}
